package com.raylu.realtime.app.template;

import com.alibaba.fastjson.JSONObject;
import com.raylu.realtime.utils.DimUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import com.raylu.realtime.utils.ThreadUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2022/1/3.
 * Copyright © 2022 lucienoz. All rights reserved.
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    private ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        executorService = ThreadUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {

        executorService.submit(
                new Runnable() {
                    @Override
                    public void run() {
                        //拼接查询条件
                        Tuple2<String, String>[] conditions = getConditions();
                        Tuple2<String, Object>[] condis = new Tuple2[conditions.length];
                        for (int i = 0; i < conditions.length; i++) {
                            try {
                                String property = BeanUtils.getProperty(input, conditions[i].f0);
                                condis[i] = Tuple2.of(conditions[i].f1, property);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                                throw new RuntimeException("");
                            } catch (InvocationTargetException e) {
                                e.printStackTrace();
                                throw new RuntimeException("");
                            } catch (NoSuchMethodException e) {
                                e.printStackTrace();
                                throw new RuntimeException("");
                            }


                        }


                        //查询维度数据
                        System.out.println("查询>>" + getTable() + ":" + Arrays.toString(condis));
                        JSONObject jsonObject = DimUtil.qurayTableByPkWithCache(getTable(), condis);
                        System.out.println("查询结果>>" + getTable() + ":" + Arrays.toString(condis) + jsonObject);
                        //组装维度数据
                        if (jsonObject != null) {
                            join(input, jsonObject);
                        }
                        //返回维度组合结果
                        resultFuture.complete(Collections.singleton(input));
                    }
                }
        );


    }


}
